package com.wanglei.project

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark04_top10 {

  def main(args: Array[String]): Unit = {
    /**
     * top10热门品类
     * 品类id -> (点击数量，下单数量，支付数量)
     */

    val conf = new SparkConf().setMaster("local[2]").setAppName("top")
    val sc = new SparkContext(conf)

    // 
    val actionRdd = sc.textFile("datas/user_visit_action.txt")


    // 定义并注册累加器
    val acc = new HotCategoryAccumulator
    sc.register(acc, "hotCategory")


    actionRdd.foreach(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          acc.add((datas(6), "click"))
        } else if (datas(8) != "null") {
          val cids = datas(8).split(",")
          cids.foreach(id => acc.add(id, "order"))
        } else if (datas(11) != "null") {
          val cids = datas(11).split(",")
          cids.foreach(id => acc.add(id, "pay"))
        }
      }
    )

    val categories = acc.value.values
    categories.toList.sortWith(
      (left, right) => {
        if (left.clickCnt > right.clickCnt) {
          true
        } else if (left.clickCnt == right.clickCnt) {

          if (left.orderCnt > right.orderCnt) {
            true
          } else if (left.orderCnt == right.orderCnt) {
            left.payCnt > right.payCnt
          } else {
            false
          }

        }

        else {
          false
        }
      }
    ).take(10).foreach(println(_))

    //
    sc.stop()
  }

}

/**
 * 自定义累加器
 * 1 继承AccumulateV2，定义范型
 * IN (品类id， 行为类型)
 * OUT mutable.Map<String, HotCategory>
 */
class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {

  val map = mutable.Map[String, HotCategory]()

  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = new HotCategoryAccumulator

  override def reset(): Unit = map.clear()

  override def add(v: (String, String)): Unit = {
    val cid = v._1
    val actionType = v._2

    val category = map.getOrElse(cid, HotCategory(cid, 0, 0, 0))
    if (actionType == "click") {
      category.clickCnt += 1
    } else if (actionType == "order") {
      category.orderCnt += 1
    } else if (actionType == "pay") {
      category.payCnt += 1
    }

    map.update(cid, category)
  }

  override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
    val map1 = this.map
    val map2 = other.value

    map2.foreach {
      case (cid, hotCategory) => {
        val category = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
        category.clickCnt += hotCategory.clickCnt
        category.orderCnt += hotCategory.orderCnt
        category.payCnt += hotCategory.payCnt
        map1.update(cid, category)
      }
    }


  }

  override def value: mutable.Map[String, HotCategory] = map

}

case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)